package com.cyh.testcoroutine

class ATest {
    /*
        协程包括挂起suspend和恢复resume
        private suspend fun getUser() {} 挂起函数，里面可以使用delay(1000)延时
        挂起函数只能在协程体内或者另外一个挂起函数内调用
        挂起和阻塞的区别：线程阻塞后，这个线程的其它代码都不会运行，比如主线程按钮按下后， 长时间不会弹起
                       协程挂起，线程中有多个协程，挂起后其它协程能正常运行，比如主线程按钮按下，马上就能弹起。

        基础设施层方式创建和启动协程：原始api使用协程
            //创建协程
            val continuation = suspend {
                1 //执行协程代码，最后一行是返回的内容
            }.createCoroutine(object: Continuation<Int> {
                override val context: CoroutineContext = EmptyCoroutineContext   //创建一个空的协程上下午
                override fun resumeWith(result: Result<Int>) {
                    println("Coroutine End: $result") //打印协程执行的结果
                }
            })
            //执行协程
            continuation.resume(Unit)

        业务框架层，上层框架的支持：

        协程调度器：所有协程必须在调度器中运行，即使它们在主线程上运行也是如此
                Dispatchers.Main:主线程运行：用于：调用suspend函数、调用UI更新、更新LiveData
                Dispatchers.IO: 非主线程运行：用于： 数据库、文件读写、网络请求处理
                Dispatchers.Default: 非主线程： 用于： 数组排序、JSON数据解析、处理差异判断、CPU密集型计算； 跟IO调度器比较取消协程的操作不一样，因为密集型取消处理会困难

        协程存在任务泄露的问题，类似内存泄露，所以需要结构化并发，即通过协程作用域来解决这个问题：
               CoroutineScope协程作用域，协程必须制定协程作用域，它会跟踪协程，取消由它启动的所有协程：下面是它的几个子类：
                     GlobalScope: 全局作用域，进程级别的，Activity和Fragment销毁，协程仍在运行
                     MainScope: 在Activity中使用，需要手动在onDestroy中取消协程
                     viewModelScope: 只能在viewModel中使用，绑定viewModel的生命周期
                     lifecycleScope: 只能在Activity和Fragment中使用，会绑定Activity和Fragment的生命周期
               MainScope使用示例：
                  val mainScope = MainScope() //定义Activity成员变量，如果MyActivity：AppCompatActivity(),CoroutineScope by MainScope(){},则这一行可以注释掉，然后下面的mainScope全部删除，可以直接调用launch{}和cancel()
                  mainScope.launch {
                        这里可以调用挂起函数，更新UI,耗时操作等等等等
                        val user = withContext(Dispatchers.IO) {
                            修改协程调度器，比如网络请求的代码在这里， 但是像Retrofit自带协程调度器，外面可以不包裹这层IO调度器
                        }
                        try {
                            delay(1000) //表示挂起1秒，这个函数只能在协程中使用
                        } catch(e: Exception) {
                            e.printStackTrace()  //挂起时，取消协程，这里能捕获到异常CancellationException
                        }

                   }  //某个函数里面调用
                   在Activity的onDestroy调用mainScope.cancel()， 如果协程还在挂起，就取消，会报异常，但这种异常不会杀死应用。

               viewModelScope使用示例：
                    在activity中onCreate中，binding.lifecycleOwner = this, 类似生命周期，这个跟viewModelScope没有关系
                    viewMode中某个函数直接调用：viewModel本身就带有viewModelScope字段，不需要new
                          viewModelScope.launch {耗时操作，网络请求等}


      launch和async都是协程作用域构建器的启动方法，区别如下：
            launch: 返回一个Job,但不附带任何结果值， 等待执行完成用join
            async: 返回一个Deferred, Deferred也是一个Job,可以使用await()在一个延期的值上得到它的最终结果, 等待执行完用await
            fun test1() = runBlocking { //这个会将主线程包装成协程作用域，里面就可以做各种协程操作了,阻塞的，会等待这个函数的所有子协程都执行完毕，主线程才会继续往下执行
                    val job1 = launch {
                        delay(1000)
                    }
                    job1.join()  //只有job1协程执行完后，才会去执行下面的协程和代码
                    val job2 = async {
                        delay(2000)
                        "result job"
                    }
                    val result = job2.await() //效果跟上面一样，只是有返回结果
            }

            fun test2() = runBlocking {
                val time = measureTimeMillis {  //用于计算执行花费的时间
                    doOne() //挂起函数，延时1秒
                    doTwo() //挂起函数，延时1秒
                }
                println("$time") //打印出花了2秒时间
            }

            fun test3() = runBlocking {
                val time = measureTimeMillis {  //用于计算执行花费的时间
                    val one = async{doOne()} //挂起函数，延时1秒
                    val two = async{doTwo()} //挂起函数，延时1秒
                    one.await()
                    two.await()
                }
                println("$time") //打印出花了1秒时间, 两个协程都运行起来，然后一起挂起是并行的
            }

            fun test4() = runBlocking {
                val time = measureTimeMillis {  //用于计算执行花费的时间
                    val one = async{doOne()}.await() //挂起函数，延时1秒并挂起等待
                    val two = async{doTwo()}.await() //挂起函数，延时1秒并挂起等待
                }
                println("$time") //打印出花了2秒时间， 是运行一个挂起一个是串行的
            }


    协程的启动模式：
        DEFAULT: 协程创建后，立即开始调度，在调度前如果取消，将直接进入取消响应状态
        ATOMIC:  协程创建后，立即开始调度，协程执行到第一个挂起点之前不响应取消
        LAZY:    只有协程被需要时，包括主动调用协程的start、join或者await等函数时才会开始调度，如果调度前被取消，那么该协程将直接进入异常结束状态
        UNDISPATCHED: 协程创建后立即在当前函数调用栈中执行，直到遇到第一个真正挂起的点
        fun test1() = runBlocking {
            val job = launch(start = CoroutineStart.DEFAULT) {
                val s = ""  //如果是DEFAULT，执行在这儿也能响应取消； 如果是ATOMIC, 运行在这里被取消，不会响应取消，必修执行到delay()时才会响应取消
                delay(2000)
                println("2222")
            }
            delay(1000)
            job.cancel() //不会出现打印就在1秒时取消掉了
        }

        fun test2() = runBlocking {
            val job = async(start = CoroutineStart.LAZY) {  //LAZY模式，这里不会马上启动
                delay(2000)
                111
            }
            val result = job.await() //LAZY模式， 这时才会开始执行协程内的代码
        }

        fun test3() = runBlocking {
            val job = async(context = Dispatchers.IO, start = CoroutineStart.UNDISPATCHED) {
                println("thread: ${Thread.currentThread().name}") //其它模式打印是IO线程，UNDISPATCHED模式打印是主线程，因为这个函数会在当前函数调用栈执行，这个test3函数是在主线程调用的，所以是主线程
            }
        }

   协程的作用域构建器：coroutineScope与runBlocking
        runBlocking是常规函数，而coroutineScope是挂起函数，它们都会等待其协程体以及所有子协程结束，
        主要区别在于runBlocking会阻塞当前线程来等待，而coroutineScope只是挂起，会释放底层线程用于其它用途。
        coroutineScope中一个协程失败了，所有其它兄弟协程也会被取消
        supervisorScope中一个协程失败了，不会影响其它兄弟协程
        fun test1() = runBlocking {
            coroutineScope {
                var job1 = launch {
                    delay(1000)
                }
                var job2 = launch {
                    delay(2000)
                }
            }
            supervisorScope {
                var job3 = launch {
                    delay(1000)
                }
                var job4 = launch {
                    delay(2000)
                }
            }
        }


    Job对象
         对每个创建的协程（launch或async）,会返回一个Job示例，该示例是协程的唯一标识，负责管理协程的生命周期，
         可以访问Job的属性： isActive、isCancelled、isCompleted 来得到当前的状态

    协程取消
         取消作用域，会取消它的子协程
         被取消的子协程并不会影响其余的兄弟协程
         协程通过抛出一个特殊的异常 CancellationException来处理取消操作
         所有kotlinx.coroutines中的挂起函数（withContext、delay等）都是可以取消的
         fun test1() = runBlocking<Unit> {
            //val job1 = GlobalScope.launch {} //GlobalScope的上下文跟runBlocking也不一样，所有runBlocking运行完，也会去取消job1, 所以job1内的打印无法打印出来
            val scope = CoroutineScope(Dispatchers.Default)
            scope.launch {
                delay(1000)
                println("job1")
            }
            scope.launch{
                delay(2000)
                println("job2")
            }
         }
         上面运行不会有打印，因为新建的CoroutineScope的上下文不是继承runBlocking的上下文，所以不会去等待scope运行，所以runBlocking执行完成后，会去取消里面其他上下文的协程。
         如果用首字母小写的coroutineScope{}的上下文就是继承runBlocking的上下文

         fun test2() = runBlocking<Unit> {
            val scope = CoroutineScope(Dispatchers.Default)
            val job1 = scope.launch {
                delay(1000)
                println("job1")
            }
            job2 = scope.launch{
                delay(2000)
                println("job2")
            }
            delay(100)
            //job1.cancel() //job1取消，不影响job2的执行
            ////job1.cancel()
            ////job1.join() //上面已经cancel了，还有必要join吗？有，上面只是通知cancel,实际还需要一段时间才能真正cancel,所以这里必须加join
            //job1.cancelAndJoin() 上两句可以合成这一句
            scope.cancel() //取消这个取消作用域的所有协程
            delay(3000) //会去等待scope执行
         }

     CPU密集型任务的取消：
        比如协程中while一直处理数据，且里面没有delay代码，则cancel无法起作用？那怎么处理呢？
           方式一：加上isActive条件，取消时，协程内部会将isActive改成false
                  while(i < 100 && isActive) {}
           方式二：加ensureActive()函数，取消时，会自动报异常，系统捕获异常，但不报错,但会退出协程
                 while(i < 100) {ensureActive()}
           方式三：加yield()函数，取消时，会自动报异常，系统捕获异常，但不报错,但会退出协程,跟上面的区别是会出让协程的执行权， 如果没有取消则yield的后面的代码会执行，每次都会出让一次执行权限
                 while(i < 100) {yield()}

   协程取消的副作用： 使用try--catch--finally
         在finally中释放资源
         use函数只能被实现了Closeable的对象使用，程序结束的时候会自动调用close方法， 常用的读取文件时需要在finally里面close文件, 而用use读取文件会自动close文件
         cancelAnJoin()后，finally里面如果还写delay(),则这个挂起不起作用，delay后面的代码也不会执行。
         finally里面怎么用delay()函数呢？
             withContext(NonCancellable) {delay()} //这样delay()前后插入的代码就都能执行了

   超时任务：
        在协程中调用：withTimeOut(1000) {delay(2000)}, 超时会报错，且进程会关闭
        如何捕获超时任务了？
            val result = withTimeoutOrNull(1000) {
                delay(2000)
                "OK"
            } 如果超时了，返回null, 否则返回OK, 不会报错


    协程上下：
        CoroutineContext是一组用于定义协程行为的元素，它由如下几项构成：
             Job: 控制协程的生命周期
             CoroutineDispatcher: 向合适的线程分发任务
             CoroutineName: 协程的名称，调试的时候很有用
             CoroutineExceptionHandler: 处理未被捕获的异常
         fun test1() = runBlocking<Unit> {
            launch(Dispatchers.Default + CoroutineName("cyh")) { //协程的上下文使用了运算符重载
                println("协程上下文加了协程名字时，打印线程名字后面会自定加上协程的名字：${Thread.currentThread().name}")
            }
         }

    协程上下文的继承：
         fun test1() = runBlocking<Unit> {
            val scope = CoroutineScope(Job() + Dispatchers.IO + CoroutineName("cyh"))  //协程作用域
            val job = scope.launch {
                println("${coroutineContext[Job]}  ${Thread.currentThread().name}")   //coroutineContext[Job]可以获取协程上下文的某一部分，每个协程的Job都是全新的，其他参数不变
                val result = async {
                    println("${coroutineContext[Job]}  ${Thread.currentThread().name}")
                    “OK”
                }.await()
            }
            job.join()
            //scope、job、result他们的协程上下文是相同的，继承关系
         }
         协程上下文 = 默认值 + 继承的CoroutineContext + 参数
         CoroutineDispatcher的默认值是：Dispatchers.Default
         CoroutineName的默认值是：coroutine
         fun test2() = runBlocking<Unit> {
            val coroutineExceptionHandler = CoroutineExceptionHandler {_, exception ->
                println("Caught $exception")
            }
            val scope = CoroutineScope(Job() + Dispatchers.Main + CoroutineName("cyh") + coroutineExceptionHandler)
            val job = scope.launch(Dispatchers.IO) {} //其他的继承scope,但是调度器被给出了IO调度器
         }


    协程异常处理：
        协程构建器的两种形式：自动传播异常（launch与actor）,向用户暴露异常（async与produce）,当这些构建器用于创建一个根协程时（即该协程不是另外一个协程的子协程），
                 自动传播异常会在它发生的第一时间被抛出，
                 而用户暴露异常则依赖用户来最终消费异常，例如通过await或receive
        fun test1() = runBlocking<Unit> {
            val job = GlobalScope.launch {
                try {
                    throw IndexOutOfBoundsException()
                }catch(e: Exception) {  //launch用于根协程中抛出的异常，必修在这里才能捕捉， 不捕获这个异常不会被抓取
                    println("Caught $e")
                }
            }
            job.join()

            val deferred = GlobalScope.async {
                throw ArithmeticException()
            }
            try {
                deferred.await()
            }catch(e: Exception) {  //async用于根协程中抛出的异常，必修在这里才能捕获， 不await()不捕获这个异常不会被抓取，进程被关闭
                println("Caught $e")
            }
        }

        非根协程的异常总是会被传播
        fun test1() = runBlocking<Unit> {
            val scope = CoroutineScope(Job())
            val job = scope.launch {
                async{
                    throw IllegalArgumentException() //不需要async.await(), 控制台也能够打印出这个异常
                }
            }
            job.join()
        }


   异常的传播特性：
        子协程出现异常，传递给它的父协程，然后父协程会取消它的所以子协程，取消自己的协程，将异常传播并传递给它的父级。
        怎样让子协程出现异常，自己处理，不传播给父协程呢？ 答案是使用SupervisorJob或者supervisorScope
        fun test1() = runBlocking<Unit> {
            val supervisor = CoroutineScope(SupervisorJob())
            val job1 = supervisor.launch {
                delay(100)
                throw IllegalArgumentException()
            }
            val job2 = supervisor.launch {
                try {
                    delay(500)
                    throw IllegalArgumentException()
                }catch(e: Exception) {
                }finally{
                }
            }
            joinAll(job1, job2)  //job1抛出了一次，控制台能打印出，但进程不会关闭，job2能正常执行，能捕获到异常,但是job1和job2外面的异常一样后关闭进程关闭所以子协程
        }

        fun test2() = runBlocking<Unit> {
            supervisorScope{
                val job1 = launch {
                    delay(100)
                    throw IllegalArgumentException()
                }
                val job2 = launch {
                    try {
                        delay(500)
                        throw IllegalArgumentException()
                    }catch(e: Exception) {
                    }finally{
                    }
                }
                joinAll(job1, job2)  //job1抛出了一次，控制台能打印出，但进程不会关闭，job2能正常执行，能捕获到异常,但是job1和job2外面的异常一样后关闭进程关闭所以子协程
            }
        }

   CoroutineExceptionHandler对协程的异常捕获，需要满足下面两个条件才会捕获：
        时机：launch里面的异常会捕获，async里面的异常不会捕获
        位置：在CoroutineScope的CoroutineContext中或者在一个根协程（CoroutineScope或supervisorScope的直接子协程）中才会捕获
        fun test1() = runBlocking<Unit> {
            val handler = CoroutineExceptionHandler {_, exception ->
                println("Caught $exception")
            }
            val job = GlobalScope.launch(handler) {
                throw AssertionError()  //这个异常可以捕获到
            }
            val deferred = GlobalScope.async(handler) {
                throw ArithmeticException()  //这个异常不能捕获到，控制台打印报错，进程关闭
            }
            job.join()
            deferred.await()
        }

        fun test2() = runBlocking<Unit> {
            val handler = CoroutineExceptionHandler {_, exception ->
                println("Caught $exception")
            }
            val scope = CoroutineScope(Job())
            val job = scope.launch(handler) {
                launch {
                    throw IllegalArgumentException()  //这个异常可以捕获到
                }
            }
            job.join()
        }

        fun test3() = runBlocking<Unit> {
            val handler = CoroutineExceptionHandler {_, exception ->
                println("Caught $exception")
            }
            val scope = CoroutineScope(Job())
            val job = scope.launch {
                launch(handler) {
                    throw IllegalArgumentException()  //这个异常不能捕获到
                }
            }
            job.join()
        }

    全局异常捕获：
        全局异常处理器可以获取到所有协程未处理的无捕获的异常，不过它不能对异常进行捕获，也不能阻止程序奔溃，但可以获得异常然后上报。
        创建src/main/resources/META-INF/services目录，新建kotlinx.coroutines.CoroutineExceptionHandler的文件。
        在MainActivity的同级目录创建下面这个文件。
        class GlobalCoroutineExceptionHandler: CoroutineExceptionHandler {
            override val key = CoroutineExceptionHandler
            override fun handlerException(context: CoroutineContext, exception: Throwable) {
                println(“可以在这里捕获所有异常，并上报”)
            }
        }


    取消与异常：
        协程取消时，会报CancellationException异常，这个异常会被忽略
        子协程取消时，不会取消它的父协程
        报CancellationException以外的异常时，会传递到父协程，父协程所有子协程结束后，父协程才会处理。

    异常聚合：
       多个子协程出现异常，一般取第一个异常进行处理，其它的异常都将绑定到第一个异常之上。
       fun test1() = runBlocking<Unit> {
            val handler = CoroutineExceptionHandler {_, exception ->
                println("Caught $exception}") //打印第一个异常
                println("Caught ${exception.suppressed.contentToString()}") //数组，打印第二个、第三个...异常
            }
            val job = GlobalScope.launch(handler) {
                launch{
                    try {
                        delay(200)
                    }finally{
                        throw ArithmeticException() //第二个异常
                    }
                }
                launch{
                    delay(100)
                    throw IOException()  //第一个异常
                }
            }
            job.join()
        }

    Flow:
        返回多个值但不是异步：集合非异步的--序列Sequence<Int>是阻塞的sequence{}多次返回多个值--挂起函数是异步返回多个值但是却是一次性返回的
        flow是异步返回多个值，但每次只返回一个值，多次返回。 flow的函数不需要suspend修饰符，使用场景，子线程下载文件，将进度来更新UI
        flow是冷流：只有在调用collect时才会去运行testFlow()里面的代码，如果多次调用collect,那么testFlow()会执行多次，并且是挂起的，即第一个collect中所有数据都接收后，才会运行后面的代码。
        fun testFlow() = flow<Int> {
            for (i in 1..3) {
                delay(1000)
                emit(i) //发射，产生一个元素
            }
        }
        fun test1() = runBlocking<Unit> {
            testFlow().collect{ value -> println(value)}
        }

        flow常见的一张使用方法：
            (1..5).asFlow().filter{
                it%2 == 0
            }.map{
                "String $it"
            }.collect{
                println("collect $it")
            }

        流的构建方式：flow<Int>{}、flowOf、asFlow()
            flowOf("one", "two").onEach{delay(1000)}.collect{value -> println(value)}

        流的收集collect必须在协程的上下文中使用, 但是流的创建则可以在普通的函数里面创建。但是流的收集和创建其实是在同一个协程上下文里面的。
        当然可以切换上下文，但是不能通过withContext(Dispatchers.IO)来切换，而应该用flowOn来切换上下文。
            fun testFlow() = flow<Int> {
                for (i in 1..3) {
                    delay(1000)
                    emit(i) //发射，产生一个元素
                }
            }.flowOn(Dispatchers.Default) //这样就实现了，创建发射在子线程中，收集在主线程中

        那如何实现流的创建发射和初步处理在子线程中执行，而收集处理在另外一个子线程中处理，这时就不能用collect收集了，而应该用launchIn收集
            fun testEvents() = (1..3).asFlow().onEach{delay(100)}.flowOn(Dispatchers.Default)
            fun test01() = runBlocking<Unit> {
                val job = testEvents().onEach{event -> println("$event--${Thread.currentThread().name}")}
                            .launchIn(CoroutineScope(Dispatchers.IO)) //.launchIn(this)这个表示当前线程，即主线程
                            join() //需要加这一句，否则不会等它执行test01()的协程就关闭了

               //job.cancelAndJoin() 流的取消方式一
               //withTimeoutOrNull(100){} 流的取消方式二，超时取消
               //testEvents().collect{value -> if(value==2) cancel()} 流的取消方式三， 但是如果testEvents()里面没有delay()或者没有emit()这两个函数，则表示密集型运算，这种取消会报错
               //testEvents().cancellable().collect(){value -> if(value==2) cancel()} 取消方式四，这种方式在密集型运算的情况下取消也不会报错
            }

         背压：
             发送速度超过收集速度时会出现背压
             解决背压能提高处理速度，没有解决背压时是发送一个处理一个,用时1.2s，解决了背压是将没有处理的缓存起来，收集有空闲时再处理缓存的数据
             fun testFlow() = flow<Int> {
                for (i in 1..3) {
                    delay(100)
                    emit(i)
                }
            }
            fun test01() = runBlocking<Unit> {
                testFlow().buffer(50).collect{value -> delay(300)} //加了buffer可以解决背压问题， 发射是串行的，用时1s
                testFlow().flowOn(Dispatchers.Default).collect{value -> delay(300)}//flowOn切换线程也能解决背压问题，因为它自带缓冲， 发射是并行的。 用时1s
                testFlow().conflate().collect{value -> delay(300)} //处理背压，但是会抛弃掉中间的值，可能中间只能打印到1，其它两个值抛弃了
                testFlow().collectLatest{value -> delay(300)} //只收集最后一个值， 只打印3
            }

        转换操作符: map、 transform
            (1..3).asFlow().map{"response ${it}"}.collect{println(it)}
            (1..3).asFlow().transform{request ->
                emit("response1: ${request}")
                emit("response2: ${request}")
            }.collect{println(it)} //打印出六个字符串

        限长操作符：take
            flow<Int> {
                try {
                    emit(1)
                    emit(2)
                    emit(3)
                } finally {
                }
            }.take(2).collect{println(it)}  //只打印1,2， emit(2)后面的代码都不会去执行

        末端操作符：collect、toList、toSet、first、single、reduce、fold
            val sum = (1..5).asFlow().map{it*it}.reduce{a,b -> a + b} ==> sum = 55

        组合操作符： zip
            val numbs = (1..3).asFlow()
            val strs = flowOf("one", "two", "three")
            numbs.zip(strs) {a, b -> "$a --> $b"}.collect{println(it)} //如果上面两句代码都加上onEach然后加延时，也能一一对应上。

        展平操作符：flatMapConcat、flatMapMerge、flatMapLatest
            fun testFlow(i:Int) = flow<String> {
                emit("$i-->aaa")
                delay(100)
                emit("$i-->bbb")
            }
            (1..3).asFlow().onEach{delay(100)}.flatMapConcat{testFlow(it)}.collect{println(it)} //打印结果1、2、3分别跟aaa、bbb相结合
            //而如果用flatMapMerge， 则分别是拿到aaa、bbb分别更1、2、3相结合。跟上面只是打印顺序不一样
            //flatMapLatest 则是拿最新的来结合

        流的异常处理：
            //在收集的地方出现异常，通过try..catch来捕获异常
                fun simpleFlow() = flow<Int> {
                    for(i in 1..3) {
                        println("Emitting $i")
                        emit(i)
                    }
                }
                fun test1() = runBlocking<Unit> {
                    try {
                        simpleFlow().collect{ value ->
                            println(value)
                            check(value <= 1) {"Collected $value"} //>1时会抛出异常
                        }
                    } catch(e: Throwable) {
                        println("Caught $e")
                    }
                }

            //在发送的地方出现异常,通过.catch{}来捕获异常
                fow {
                    emit(1)
                    throw ArithmeticException("Div 0")
                }.catch{e:Throwable -> println("Caught $e")} //捕获到异常可以在里面调用emit(2)
                .flowOn(Dispatchers.IO)
                .collect{println(it)}

        流的完成：finally只知道已完成、onCompletion不仅知道正常的已完成，还知道如果是报错导致的已完成还可以获得报错信息，不仅可以捕获发送端的异常，还可以捕获接收端的异常
            try {
                flow { emit(i) }.collect{println(it)}
            }finally{
                println("已完成--方式一")
            }

            flow { emit(i) }.onCompletion{println("已完成--方式二")}.collect{println(it)}

            fow {
                    emit(1)
                    throw ArithmeticException("Div 0")
                }
                .onCompletion{e ->
                    if(e != null) {
                        println(e)
                    } else {
                        print("已完成")
                    }
                }
                .collect{println(it)}



   Channel 通道，实际是一个并发安全的队列，用来连接协程，实现不同协程的通信
        生产者发送协程到Channel,消费者接收Channel里面的协程
        Channel是一个热流
        fun test01() = runBlocking<Unit> {
            val channel = Channel<Int>()
            //生产者
            val producer = GlobalScope.launch {
                var i = 0
                while (true) {
                    delay(1000)
                    channel.send(++i)
                    println("send $i")
                }
            }
            //消费者
            val consumer = GlobalScope.launch {
                while (true) {
                    val element = channel.receive()
                    println("receive $element")
                }
            }
            joinAll(producer, consumer)
        }
        channel有一定的缓冲区，如果缓冲区满了，还没有receive取走，send就会挂起，即等在那里不发送，直到receive取走数据后才恢复。默认缓冲区大小是0，即每次只能发一个然后挂起等待接收

        fun test02() = runBlocking<Unit> {
            val channel = Channel<Int>(Channel.UNLIMITED)  //设置缓冲区大小为不限制，即无限多个
            //生产者
            val producer = GlobalScope.launch {
                for(x in 1..5) {
                    channel.send(x * x)
                    println("send ${x * x}")
                }
            }
            //消费者
            val consumer = GlobalScope.launch {
                val iterator = channel.iterator()
                while(iterator.hasNext()) {          //或者 for(element in channel){}
                    val element = iterator.next()
                    println("receive $element")
                    delay(2000)
                }
            }
            joinAll(producer, consumer)
        }
        上面这种方式是设置缓冲区大小为无限大，然后全部发生到缓冲区，然后再一个个接收

        produce与actor是构造生产者和消费者的便捷方式，都能启动一个协程
            fun test03() = runBlocking<Unit> {
                val receiveChannel:ReceiveChannel<Int> = GlobalScope.produce{
                    repeat(100) {
                        delay(1000)
                        send(it)
                    }
                }
                val consumer = GlobalScope.launch {
                    for (i in receiveChannel) {
                        println("received: $i")
                    }
                }
                consumer.join()
            }

            fun test04() = runBlocking<Unit> {
                val sendChannel: SendChannel<Int> = GlobalScope.actor<Int> {
                    while(true) {
                        val element = receive()
                        println("received: $element")
                    }
                }
                val producer = GlobalScope.launch{
                    for (i in 1..3) {
                        sendChannel.send(i)
                    }
                }
                producer.join()
            }

       Channel的关闭，对应的协程关闭，channel也会跟着关闭
            channel.close()
            关闭表示生产者不再生产新的元素，消费者还可以继续消化缓冲区存在的元素
            调用close()后，isClosedForSend是true, 所有元素读取后，isClosedForReceive返回true
            Channel的生命周期最好由主导方来维护，建议由主导的一方实现关闭
            fun test01() = runBlocking<Unit> {
                    val channel = Channel<Int>(3)  //设置缓冲区大小为不限制，即无限多个
                    //生产者
                    val producer = GlobalScope.launch {
                        List(3) {
                            channel.send(it)
                            println("send $it")
                        }
                        channel.close()
                        println("isClosedForSend = ${channel.isClosedForSend} --- isClosedForReceive = ${channel.isClosedForReceive}")
                    }
                    //消费者
                    val consumer = GlobalScope.launch {

                        for(element in channel){
                            println("receive $element")
                            delay(1000)
                        }
                        println("isClosedForSend = ${channel.isClosedForSend} --- isClosedForReceive = ${channel.isClosedForReceive}")
                    }
                    joinAll(producer, consumer)
            }


       BroadcastChannel:
         Channel一个发送者可能有多个接收者，但是如果一个接收者接收了，其它的接收者将无法接收到这次的发送，如果想要其它的接收者也接收到，可以使用BroadcastChannel
         fun test01() = runBlocking<Unit> {
            val broadcastChannel = BroadCastChannel<Int>(Channel.BUFFERED)
            val producer = GlobalScope.launch {
                List(3) {
                    delay(100)
                    broadcastChannel.send(it)
                }
                broadcastChannel.close()
            }
            List(3) { index ->
                GlobalScope.launch {
                    val receiveChannel = broadcastChannel.openSubscription()
                    for(i in receiveChannel) {
                        println("[#$index] received: $i")
                    }
                }
            }.joinAll()
         }

         channel和BroadcastChannel可以相互转换
            val channel = Channel<Int>()
            val broadcastChannel = channel.broadcast(3)


      多路复用：Multiplexing
           一个信道同时传递多路信号， 比如从后台两个接口获取数据，先拿到哪个就用哪个，另外一个丢弃。
           使用select()和onWait()来实现
           fun test01() = runBlocking<Unit> {
                val channels = listOf(Channel<Int>(), Channel<Int>())
                GlobalScope.launch{
                    delay(100)
                    channels[0].send(1)
                }
                GlobalScope.launch{
                    delay(50)
                    channels[1].send(2)
                }
                val result = select<Int?> {
                    channels.forEach{ channel ->
                           channel.onReceive{it}
                    }
                }
                println(result)
           }

     协程不安全的并发访问
        fun test01() = runBlocking<Unit> {
            var count = 0
            List(1000) {
                GlobalScope.launch{count++}
            }.joinAll()
            println(count) //每次的值不一样，不是1000，在一个线程里面同时启动多个协程，他们的并发是不安全的。
        }
        fun test02() = runBlocking<Unit> {
            var count = AtomicInteger(0)
            List(1000) {
                GlobalScope.launch{count.incrementAndGet()}
            }.joinAll()
            println(count) //是1000， 使用原子性操作解决并发安全的问题
        }
        协程本身也提供了下面的并发安全工具：
            Channel: 使用通道是线程安全的
            Mutex: 轻量级锁，他的lock和unlock与线程锁比较类似，不会阻塞线程而是挂起
            Semaphore: 轻量级信号量， 参数为1时等价于Mutex
            fun test03() = runBlocking<Unit> {
                var count = 0
                val mutex = Mutex()
                List(1000) {
                    GlobalScope.launch{
                        mutex.withLock{
                            count++
                        }
                    }
                }.joinAll()
                println(count) //1000, 并发安全
            }

            fun test04() = runBlocking<Unit> {
                var count = 0
                val semaphore = Semaphore(1)
                List(1000) {
                    GlobalScope.launch{
                        semaphore.withPermit{
                            count++
                        }
                    }
                }.joinAll()
                println(count) //1000, 并发安全
            }

            //可以使用操作符来避免并发操作， 即count不放到协程里面
            fun test05() = runBlocking<Unit> {
                var count = 0
                var result = count + List(1000) {
                    GlobalScope.async{1}
                }.map{it.await()}.sum()
                println(result) //1000,避免并发
            }

    项目实战：
        实战一、Flow下载文件
            sealed class DownloadStatus {
                object None : DownloadStatus()
                data class Progress(val value: Int) : DownloadStatus()
                data class Error(val throwable: Throwable) : DownloadStatus()
                data class Done(val file: File) : DownloadStatus()
            }
            object DownloadManager {
                fun download(url:String, file :File) : Flow<DownloadStatus> {
                    return flow {
                        val request = Request.Builder().url(url).get.build()
                        val response = OkHttpClient.Builder().build().newCall(request).execute()
                        if(response.isSuccessful) {
                            response.body()!!.let { body ->
                                val total = body.contentLength()
                                file.outputStream().use { output ->
                                     val input = body.byteStream()
                                     var emittedProgress = 0L
                                     写文件操作后
                                     emit(DownloadStatus.Progress(progress))
                                }
                            }
                            emit(DownloadStates.Done(file))
                        } else {
                            emit(DownloadStatus.Error(it))
                        }
                    }.catch {
                        file.delete()
                        emit(DownloadStatus.Error(it))
                    }.flowOn(Dispatcher.IO)
                }
            }
            Activity中调用
                lifecycleScope.launchWhenCreated {
                    context?.apply{
                        DownloadManager.download(url, file).collect { status ->
                            when(status) {
                                is DownloadStatus.Progress -> {
                                   因为上下文在主线程，所以这里是主线程
                                }
                            }

                        }
                    }
                }


        实战二：Flow与Room的应用
            UserDao文件中
                @Insert(onConflict = OnConflictStrategy.REPLACE)
                suspend fun insert(user: User)  //自带支持协程，需要加suspend

                @Query("SELECT * FROM user")
                fun getAll(): Flow<List<User>>  //不需要加suspend
            ViewModel中使用
                fun insert() {
                    viewModelScope.launch {
                        AppDatabase.getInstance(getApplication()).userDao().insert(User())
                    }
                }

                fun getAll() : Flow<List<User>> {
                    return AppDatabase.getInstance(getApplication()).userDao().getAll().catch{e -> e.printStackTrace()}.flowOn(Dispatchers.IO)
                }

            Activity中使用：
                lifecycleScope.launchWhenCreated {
                    viewModel.getAll().collect {value ->
                        这个value就是List<User>的列表
                    }
                }


      实战三、Flow与Retrofit的应用
            @GET("article")
            suspend fun searchArticles(@Query("key") key : String):List<Article>
            viewModel文件中：
                fun searchArticles(key:String) = flow {
                    val list = searchArticles(key)
                    emit(list)
                }.flowOn(Dispatchers.IO)
                .catch{e -> e.printStackTrace()}

                或者
                val articles = MutableLiveData<List<Article>>() //监听这个数据的变化
                viewModelScope.launch{
                    上面flow的内容.collect{
                        articles.setValue(it)
                    }
                }
            Activity文件中：
                lifecycleScope.launchWhenCreate{
                    viewModel.searchArticles("aa").collect{}
                }
                或者监听articles的变化

     Flow是冷流，即Flow订阅collect之后，发射出来的值才会存在内存中，类似懒加载
     StateFlow和SharedFlow是热流，在垃圾回收之前，都是存在内存之中，并且处于活跃状态
     StateFlow类似liveData
            viewModel文件中
                val selected = MutableStateFlow<Boolean>(false)
                selected.value = value
            Activity文件中
                lifecycleScope.launch {
                    viewModel.selected.collect {
                        // ... 引起UI发生的变化
                    }
                }
     SharedFlow类似broadcastChannel
            viewModel文件中
                val sharedFlow = MutableSharedFlow<Int>(
                    5 // 参数一：当新的订阅者Collect时，发送几个已经发送过的数据给它
                    , 3 // 参数二：减去replay，MutableSharedFlow还缓存多少数据
                    , BufferOverflow.DROP_OLDEST // 参数三：缓存策略，三种 丢掉最新值、丢掉最旧值和挂起
                )
                // 初始化时调用
                init {
                    for (i in 0..10) {
                        sharedFlow.tryEmit(i)
                    }
                }

                // 在按钮中调用
                fun doAsClick() {
                    for (i in 11..20) {
                        sharedFlow.tryEmit(i)
                    }
                }
           Activity文件：
                // 启动第一个协程，接收初始化的数据,可以有多个这样的协程接收者
                lifecycleScope.launch {
                    val sb = StringBuffer()
                    viewModel.sharedFlow.collect {
                        sb.append("<<${it}")
                        tvContent.text = sb
                    }
                }


    实战四、Flow和Paging的使用
            使用流程：Repository(数据源：PagingSource) --> ViewModel(Pager与Flow<PagingData>) --> UI(PagingDataAdapter)
            MoviePagingSource文件中：
                class MoviePagingSource : PagingSource<Int, Movie>() {
                    override suspend fun load() {
                        val movies = RetrofitClient.createApi().getMovies()
                        return LoadResult.Page(
                            data = movies,
                            prevKey = ..  //这个几个参数要小心，如果当前页是1，则prevKey是null,nextKey是4，因为第一次加载了三页，当前为2时，next可以是5，以此类推
                            nextKey = ..
                        )
                    }
                }
            viewModel文件中：
                fun loadMovie():Flow<PagingData<Movie>> {
                    return Pager(
                        config = PagingConfig(
                                pageSize=8,   //如果只设置pageSize会出现问题，必须设置第一次请求的大小
                                initialLoadSize = 8 * 3  //至少是8的两倍
                            ),
                        pagingSourceFactory = {MoviePagingSource}
                    ).flow.flowOn(...)
                    上面这样写，viewModel中的数据不会缓存，如果需要缓存改成 Pager(...).flow.cachedIn(viewModelScope)
                }
            MovieAdapter文件：
                class MovieAdapter(private val context: Context) : PageDataAdapter<Movie, BindingViewHolder>() {}
            需要加载更多的图标，看不到，需要延迟几秒钟才能看到，或者网络不好时才可以看到
                movieAdapter.withLoadStateFooter()
            需要加载完成图标。。
            下拉更新图标显示和隐藏
            下拉更新操作，需要包裹一层SwipeRefreshLayout
                取消更新图标： movieAdapter.loadStateFlow.collectLatest{}  监听他的各种状态，来取消更新图标
                上拉刷新有一个问题：即设置如下：pageSize=8,initialLoadSize=8,prefetchDistance=1表示这一页的最后一行显示时加载下一页， 这样配置参数时，上拉刷新会出错。
                          解决方法：将initialLoadSize或者prefetchDistance改成更大的值才能解决这个问题


    * */
}